package com.taimanetworks.kafka;

import com.taimanetworks.kafka.parseTsp.Sourcedata;
import com.taimanetworks.utils.ClickUtils;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class ClickhouseFunction extends RichSinkFunction<Sourcedata> {
    private   List<Sourcedata> list = new CopyOnWriteArrayList<>();
    Connection connection = null;

    String sql;

    public ClickhouseFunction(String sql) {
        this.sql = sql;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = ClickUtils.getConnection();
//        myListState = getRuntimeContext().getListState(new ListStateDescriptor<Sourcedata>("my-list", Sourcedata.class));
//        connection.setAutoCommit(false);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void invoke(Sourcedata  data, Context context) throws Exception {
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        list.add(data);
        if(list.size()>10000){
        for (Sourcedata value : list) {
            preparedStatement.setString(1, value.getVin());
            preparedStatement.setString(2, value.getGpss_time());
            preparedStatement.setString(3, value.getGpss_type());
            preparedStatement.setString(4, value.getCar_statue());
            preparedStatement.setString(5, value.getRecharge_statue());
            preparedStatement.setString(6, value.getRun_statue());
            preparedStatement.setString(7, value.getVehspeed());
            preparedStatement.setString(8, value.getMilleage());
            preparedStatement.setString(9, value.getTotal_voltage());
            preparedStatement.setString(10, value.getTotal_current());
            preparedStatement.setString(11, value.getSoc());
            preparedStatement.setString(12, value.getDc_statue());
            preparedStatement.setString(13, value.getGears());
            preparedStatement.setString(14, value.getInsulation_resistance());
            preparedStatement.setString(15, value.getAccele_pedal());
            preparedStatement.setString(16, value.getBrake_paddle());
            preparedStatement.setString(17, value.getDrive_motor_control_temp());
            preparedStatement.setString(18, value.getDrive_motor_id());
            preparedStatement.setString(19, value.getDrive_motor_speed());
            preparedStatement.setString(20, value.getDrive_motor_statu());
            preparedStatement.setString(21, value.getDrive_motor_temp());
            preparedStatement.setString(22, value.getDrive_motor_torque());
            preparedStatement.setString(23, value.getMotor_bus_current());
            preparedStatement.setString(24, value.getMotor_input_voltage());
            preparedStatement.setString(25, value.getEngine_status());
            preparedStatement.setString(26, value.getCrankshaft_speed());
            preparedStatement.setString(27, value.getFuel_consume());
            preparedStatement.setString(28, value.getLatitude());
            preparedStatement.setString(29, value.getLocation_status());
            preparedStatement.setString(30, value.getLongitude());
            preparedStatement.setString(31, value.getBattery_monomer_max());
            preparedStatement.setString(32, value.getBattery_monomer_min());
            preparedStatement.setString(33, value.getMax_temp());
            preparedStatement.setString(34, value.getMax_temp_probeid());
            preparedStatement.setString(35, value.getMax_temp_systemid());
            preparedStatement.setString(36, value.getMin_temp());
            preparedStatement.setString(37, value.getMin_temp_probeid());
            preparedStatement.setString(38, value.getMin_temp_systemid());
            preparedStatement.setString(39, value.getVoltage_battery_monomerid_max());
            preparedStatement.setString(40, value.getVoltage_battery_monomerid_min());
            preparedStatement.setString(41, value.getVoltage_battery_systemid_max());
            preparedStatement.setString(42, value.getVoltage_battery_systemid_min());

            preparedStatement.setString(43, value.getBattery_cell_lowconsis_alarm());
            preparedStatement.setString(44, value.getBattery_cell_overpress_alarm());
            preparedStatement.setString(45, value.getBattery_cell_underpress_alarm());
            preparedStatement.setString(46, value.getBattery_hightemp_alarm());
            preparedStatement.setString(47, value.getBrake_system_alarm());
            preparedStatement.setString(48, value.getCommon_alarm());
            preparedStatement.setString(49, value.getDc_statu_alarm());
            preparedStatement.setString(50, value.getDc_temp_alarm());
            preparedStatement.setString(51, value.getDrive_motor_temp_alarm());
            preparedStatement.setString(52, value.getDriver_conttemp_alarm());
            preparedStatement.setString(53, value.getDriver_machine_trouble_list());
            preparedStatement.setString(54, value.getDriver_machine_trouble_total());
            preparedStatement.setString(55, value.getEngine_trouble_list());
            preparedStatement.setString(56, value.getEngine_trouble_total());
            preparedStatement.setString(57, value.getHigh_pressure_interlock_alrm());
            preparedStatement.setString(58, value.getHigh_soc_alarm());
            preparedStatement.setString(59, value.getInsulation_alarm());
            preparedStatement.setString(60, value.getLow_soc_alarm());
            preparedStatement.setString(61, value.getMax_alarm_level());
            preparedStatement.setString(62, value.getOther_trouble_list());
            preparedStatement.setString(63, value.getOther_trouble_total());
            preparedStatement.setString(64, value.getRecharge_energy());
            preparedStatement.setString(65, value.getRecharge_fault_list());
            preparedStatement.setString(66, value.getRecharge_fault_total());
            preparedStatement.setString(67, value.getSoc_hop_alarm());
            preparedStatement.setString(68, value.getTemp_diff_alarm());
            preparedStatement.setString(69, value.getVehicle_mounted_overpress_alarm());
            preparedStatement.setString(70, value.getVehicle_mounted_storahe_overcharge());
            preparedStatement.setString(71, value.getVehicle_mounted_underpress_alarm());
            preparedStatement.setString(72, value.getBattery_cell_total());
            preparedStatement.setString(73, value.getBattery_cell_voltage());
            preparedStatement.setString(74, value.getRecharge_storage_current());
            preparedStatement.setString(75, value.getRecharge_storage_voltage());
            preparedStatement.setString(76, value.getStart_battery_cell());
            preparedStatement.setString(77, value.getStart_battery_id());
            preparedStatement.setString(78, value.getSystemid());
            preparedStatement.setString(79, value.getYears());
            preparedStatement.setString(80, value.getMonths());
            preparedStatement.setString(81, value.getDays());

            preparedStatement.addBatch();

        }
            list = new CopyOnWriteArrayList<>();


        long startTime = System.currentTimeMillis();
        int[] ints = preparedStatement.executeBatch();
//        connection.commit();
//        preparedStatement.close();
        long endTime = System.currentTimeMillis();
        System.out.println("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据 = " + ints.length);
        System.out.println("再次进行初始化。。。");
        }

    }

}
